package com.bwie.article.consumer;

import com.alibaba.fastjson.JSONObject;
import com.bwie.article.mapper.ArticleMapper;
import com.bwie.common.domain.Article;
import lombok.extern.log4j.Log4j2;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

@Component
@Log4j2
public class ArticleConsumer {

    @Autowired
    private StringRedisTemplate redisTemplate;
    @Autowired
    private ArticleMapper articleMapper;

    @KafkaListener(topics = "aa",containerFactory = "kafkaListenerContainerFactory",errorHandler ="myKafkaListenerErrorHandler" )
    public void add(ConsumerRecord consumerRecord, Acknowledgment acknowledgment){
        Long count = redisTemplate.opsForSet().add("aa", (String) consumerRecord.key());
        Article article1 = (Article) consumerRecord.value();
        log.info("article",article1);
        if(count>0){
            articleMapper.addArticle(article1);
        }
        acknowledgment.acknowledge();
    }

}
